package com.atguigu.gmall.realtime.utils;

import com.atguigu.gmall.realtime.bean.TransientSink;
import com.atguigu.gmall.realtime.common.GmallConfig;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.lang.reflect.Field;
import java.sql.PreparedStatement;
import java.sql.SQLException;

//将数据写入到clickhouse的工具类
public class ClickHouseUtil {
    //获取针对clicksink的jdbcsink
    public static <T>SinkFunction<T> getJdbcSink(String sql){
        SinkFunction<T> sinkFunction = JdbcSink.<T>sink(
                sql,
                new JdbcStatementBuilder<T>() {
                    @Override
                    public void accept(PreparedStatement ps, T obj) throws SQLException {
                        /*获取当前对象的所以属性,通过反射获取属性,因为现在不知到此类是什么，属性有什么，所以需要通过反思获取
                        getFields()：获得某个类的所有的公共（public）的字段，包括父类中的字段。
                        getDeclaredFields()：获得某个类的所有声明的字段，即包括public、private和proteced，但是不包括父类的申明字段*/
                        Field[] fields = obj.getClass().getDeclaredFields();
                        //对属性进行遍历,然后写入到clickhouse中
                        //定义一个变量，用来计算跳过的次数
                        int jumpNum = 0;
                        for (int i = 0; i < fields.length; i++) {
                            //得到没一个属性
                            Field field = fields[i];
                            //判断在属性上面是否加入跳过的注解
                            TransientSink transientSink = field.getAnnotation(TransientSink.class);
                            //如果transientSink不为空的话，就是跳出的属性，此时我们需要进行计数,然后跳出此次循环
                            if (transientSink != null) {
                                jumpNum++;
                                System.out.println("跳过属性:" + field.getName());
                                continue;
                            }
                            //设置私有属性的访问权限
                            field.setAccessible(true);
                            //获取属性的值,然后将获取到的值赋值给占位符
                            try {
                                Object fieldValue = field.get(obj);
                                ps.setObject(i + 1 - jumpNum, fieldValue);
                            } catch (IllegalAccessException e) {
                                e.printStackTrace();
                                new RuntimeException("给占位符赋值失败");
                            }
                        }
                    }
                },
                new JdbcExecutionOptions.Builder().withBatchSize(5).build(),
                new JdbcConnectionOptions
                        .JdbcConnectionOptionsBuilder()
                        .withUrl(GmallConfig.CLICKHOUSE_SERVER)
                        .withDriverName("ru.yandex.clickhouse.ClickHouseDriver")
                        .build()


        );
        return sinkFunction;
    }

}
